module share.nettcp;


import core.thread, core.sync.mutex;
import std.socket, std.string, std.conv, std.concurrency, std.parallelism;
import share.container, share.streambf;


const TCP_MAXBUF = 1024 * 4;
const TCP_MAX_SEND_CHUNK = 1024 * 2;
const TCP_THREAD_LOOP_TERM = 40;

const TCP_DEF_TIMEOUT = 30000;

const TCP_UNINIT = 0;
const TCP_INIT = 1;
const TCP_WAITING_THREAD = 11;

const TCP_DISCONNECT = 0;
const TCP_CONNECT_TRYING = 1;
const TCP_CONNECT = 2;
const TCP_RECV = 3;
const TCP_ERR_CONNECT = 21;
const TCP_ERR_SEND = 22;
const TCP_ERR_RECV = 23;


struct TcpPacket
{
    public void[] _buffer = null;
    public size_t _buflen = 0;

    public this( size_t buflen )
    {
        expand( buflen );

        if (_buffer == null)
            throw new Exception( "TcpPacket::TcpPacket cannot allocate memory!" );
    }
    
    public ~this()
    {
        if ( _buffer !is null )
            _buffer = null;
    }

    public bool expand( size_t buflen )
    {
        if ( _buflen < buflen )
        {
            _buffer.length = buflen;
            _buflen = buflen;

            return true;
        }

        return false;
    }



    //operator const char * const ()    { return (const char *) m_pBuf; }
    
    @property public void[] data() { return _buffer; }
    @property public size_t size() const { return _buflen; }
}
alias TcpPacket* PTcpPacket;



class TcpHandler : Mutex
{
    public byte m_nInstanceState;
    public NetIoThread m_hNetIoThread;
    
    public uint m_nMaxTimeOut;
    
    public ushort _state;
    public Socket _sock;

    public Queue!PTcpPacket m_qSend;
    public PTcpPacket m_pOutputPacket;
    public uint m_nOutputBytes;
    
    public StreamBuffer!(ubyte) m_qRecv;
    
    public this()
    {
        super();
        
        m_nInstanceState = TCP_UNINIT;
        m_hNetIoThread = null;
        
        m_nMaxTimeOut = 0;
        
        _state = TCP_DISCONNECT;
        _sock = null;
        
        m_qSend = m_qSend.init;
        m_pOutputPacket = m_pOutputPacket.init;
        m_nOutputBytes = 0;

        m_qRecv.Expand(TCP_MAXBUF);
    }
    
    public ~this()
    {
        Reset();
    }
    
    public void Reset()
    {
        if ( m_hNetIoThread !is null )
        {
            m_nInstanceState = TCP_WAITING_THREAD;

            if ( m_hNetIoThread.isRunning() )
                m_hNetIoThread.join();

            m_hNetIoThread = null;
        }

        lock();
        scope(exit) unlock();

        if ( _sock !is null )
        {
            if ( _sock.handle() != socket_t.init )
            {
                _sock.shutdown(SocketShutdown.BOTH);
                _sock.close();
            }
            _sock = null;
        }

        m_pOutputPacket = null;

        _state = TCP_DISCONNECT;
        m_nInstanceState = TCP_UNINIT;

        m_qRecv.ClearAll();
        m_qSend.clear();
    }
    
    public bool Connect( InternetAddress addr, uint nMaxTimeOut = TCP_DEF_TIMEOUT )
    {
        try
        {
            _sock = new TcpSocket(AddressFamily.INET);
            _sock.blocking(false);
            _sock.setOption(SocketOptionLevel.SOCKET, SocketOption.SNDTIMEO, nMaxTimeOut);
            _sock.setOption(SocketOptionLevel.SOCKET, SocketOption.RCVTIMEO, nMaxTimeOut);
            _sock.connect(addr);
        }
        catch (SocketException se)
        {
            //if ( wouldHaveBlocked() == true )

            return false;
        }
        
        m_nMaxTimeOut = nMaxTimeOut;
        
        m_nInstanceState = TCP_INIT;
        _state = TCP_CONNECT_TRYING;

        
        try
        {
            m_hNetIoThread = new NetIoThread(this);
            m_hNetIoThread.start();
        }
        catch (ThreadException ex)
        {
            throw ex;
        }
        
        return true;
    }
    
    public bool Connect( in string addr, ushort port, uint nMaxTimeOut = TCP_DEF_TIMEOUT )
    {
        return Connect( new InternetAddress( addr, port ), nMaxTimeOut );
    }
    
    public void Disconnect()
    {
        Reset();
    }
    
    /*
    Poll()

    ÀÌ ÇÔ¼öÀÇ »óÅÂ °ªÀÌ TCP_RECV¶ó¸é, »ç¿ëÀÚ°¡ ÇÒ´çµÈ ¸Þ¸ð¸®¸¦ Á÷Á¢ ÇØÁ¦ÇØ¾ß ÇÑ´Ù.

    »ç¿ë ¿¹>
    ---------------------------------------------------------------
    ushort      nState;
    CTcpPacket  *pPacket;

    while ( Poll( &nState, &pPacket ) )
    {
        if ( nState == TCP_RECV )
        {
            ProcessRecvPacket();
            delete pPacket;  <-----
        }
    }
    ---------------------------------------------------------------
    */
    public bool Poll( ref ushort state, ref PTcpPacket packet )
    {
        lock();
        scope(exit) unlock();

        state = 0;

        /*int copy_count = m_qRecv.size();
        if ( copy_count > 0 &&
            CopyCompletionPacket( m_qRecv.buffer(), copy_count, packet ) )
        {
            m_qRecv.Remove( copy_count );
            state = TCP_RECV;
            
            return true;
        }*/

        if ( m_qRecv.size() > 0 &&
            CopyCompletionPacket( m_qRecv.buffer(), m_qRecv.size(), packet ) )
        {
            m_qRecv.Remove( packet.size() );
            state = TCP_RECV;
            
            return true;
        }
        
        if ( _state > 0 && _state != TCP_RECV )
        {
            state = _state;
            
            if ( _state == TCP_CONNECT )
                _state = 0;
            
            return true;
        }
        
        return false;
    }
    
    public void Send( void[] buffer, int buflen )
    {
        auto packet = new TcpPacket( buflen );
        if ( packet is null )
            throw new Exception( "TcpHandler::Send Could not allocate a TcpPacket." );

        packet._buffer[0 .. buflen] = buffer[0 .. buflen];

        lock();
        m_qSend.enqueue(packet);
        unlock();
    }
    
    protected bool CopyCompletionPacket( void[] buffer, int buflen, ref PTcpPacket packet )
    {
        packet = new TcpPacket( buflen );
        if ( packet is null )
            throw new Exception( "TcpHandler::CopyCompletionPacket Could not allocate a TcpPacket." );
        //packet.expand(buflen);

        packet._buffer[0 .. buflen] = buffer[0 .. buflen];

        return true;
    }
    
    protected void ProcessInput()
    {
        // recv() ÀÎÀÚ
        ubyte buf[TCP_MAXBUF];
        int nRecvLen;
        
        while ( true )
        {
            nRecvLen = _sock.receive( buf, SocketFlags.NONE );
            if ( nRecvLen == 0 )
            {
                _state = TCP_DISCONNECT;
                break;
            }
            else if ( nRecvLen < 0 )
            {
                if ( wouldHaveBlocked() == false )
                    _state = TCP_ERR_RECV;
                
                break;
            }

            m_qRecv.Append( buf, nRecvLen );
        }
    }
    
    protected void ProcessOutput()
    {
        // TCP ¿¬°áÀÌ ÀÌ·ç¾îÁø ½ÃÁ¡ºÎÅÍ Ãâ·ÂÀÌ °¡´ÉÇØÁø´Ù.
        if ( _state == TCP_CONNECT_TRYING )
            _state = TCP_CONNECT;
        
        // send() ÀÎÀÚ
        int nSendLen, nRemain, nToSend;

        if ( m_pOutputPacket is null )
        {
            if ( !m_qSend.empty() )
            {
                m_pOutputPacket = m_qSend.dequeue();
            }
        }

        while ( m_pOutputPacket !is null )
        {
            nRemain = m_pOutputPacket.size() - m_nOutputBytes;
            nToSend = std.algorithm.min(nRemain, TCP_MAX_SEND_CHUNK);

            auto slice = m_pOutputPacket.data()[m_nOutputBytes .. m_nOutputBytes + nToSend];

            nSendLen = _sock.send( slice, SocketFlags.NONE );

            if ( nSendLen == 0 )
            {
                _state = TCP_DISCONNECT;
                break;
            }
            else if ( nSendLen < 0 )
            {
                if ( wouldHaveBlocked() == false )
                    _state = TCP_ERR_SEND;
                
                break;
            }
            
            m_nOutputBytes += nSendLen;
            
            if ( m_nOutputBytes == m_pOutputPacket.size() )
            {
                m_nOutputBytes = 0;
                m_pOutputPacket = null;

                if ( !m_qSend.empty() )
                    m_pOutputPacket = m_qSend.dequeue();
            }
        }
    }
    
    protected void ProcessExcept()
    {
        _state = TCP_ERR_CONNECT;
    }
    
    
}

class NetIoThread : Thread 
{
    private TcpHandler _handler;

    public this(TcpHandler handler)
    {
        _handler = handler;
        super( &execute );
    }

    private void execute()
    {
        if (_handler is null) return;
        
        // select() ÀÎÀÚ
        SocketSet readfds = new SocketSet;
        SocketSet writefds = new SocketSet;
        SocketSet exceptfds = new SocketSet;
        TimeVal timeout;
        
        while (true)
        {
            if (_handler.m_nInstanceState == TCP_WAITING_THREAD)
                break;
            
            Thread.sleep( dur!("msecs")( TCP_THREAD_LOOP_TERM ) );
            
            readfds.reset();
            writefds.reset();
            exceptfds.reset();
            
            readfds.add(_handler._sock);
            writefds.add(_handler._sock);
            exceptfds.add(_handler._sock);
            
            if (_handler._sock is null)
                return;

            timeout.seconds = 0;
            timeout.microseconds = 0;
            if ( Socket.select( readfds, writefds, exceptfds, &timeout ) <= 0 )
                continue;
            
            _handler.lock();
            
            if (readfds.isSet(_handler._sock))
                _handler.ProcessInput();
            
            if (writefds.isSet(_handler._sock))
                _handler.ProcessOutput();
            
            if (exceptfds.isSet(_handler._sock))
                _handler.ProcessExcept();
            
            _handler.unlock();
        }
    }
}
